package com.doit.demo.day06;

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.util.ArrayList;
import java.util.List;

/**
 * @DATE 2022/2/20/16:28
 * @Author MDK
 * @Version 2021.2.2
 **/
public class ListStateDemo {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.enableCheckpointing(5000);
        //将同一个用户最近的10个行为保存起来
        //u001,view
        //u002,view
        //u001,pay
        //u001,addCart
        //u002,view
        //u002,addCart
        DataStreamSource<String> lines = env.socketTextStream("linux01", 8888,"\n",5);

        SingleOutputStreamOperator<Tuple2<String, String>> tpStream = lines.map(new MapFunction<String, Tuple2<String, String>>() {
            @Override
            public Tuple2<String, String> map(String value) throws Exception {
                if (value.startsWith("error")) {
                    throw new RuntimeException("运行出错了!!!");
                }
                String[] fields = value.split(",");
                String uid = fields[0];
                String event = fields[1];
                return Tuple2.of(uid, event);
            }
        });

        KeyedStream<Tuple2<String, String>, String> keyedStream = tpStream.keyBy(t -> t.f0);
        SingleOutputStreamOperator<Tuple2<String, List<String>>> res = keyedStream.map(new UserEventFunction());

        res.print();
        env.execute();
    }


    private static class UserEventFunction extends RichMapFunction<Tuple2<String, String>, Tuple2<String, List<String>>>{

        private ListState<String> listState;

        @Override
        public void open(Configuration parameters) throws Exception {
            //定义状态描述器
            ListStateDescriptor<String> stateDescriptor = new ListStateDescriptor<>("event-state", String.class);
            //初始化或恢复状态
            listState = getRuntimeContext().getListState(stateDescriptor);
        }

        @Override
        public Tuple2<String, List<String>> map(Tuple2<String, String> input) throws Exception {
            String event = input.f1;
            listState.add(event);
            ArrayList<String> events = (ArrayList<String>)listState.get();
            if(events.size() > 10){
                events.remove(0);
            }
            return Tuple2.of(input.f0, events);
        }
    }
}
